package com.atguigu.Flink.sql.connector;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Flink01_DataGenConnector {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //基于DataGen connector创建连接器表
        String createSourceSql =
                " create table t1 (" +
                        " id STRING , " +
                        " vc INT , " +
                        " ts BIGINT " +
                        " ) WITH (" +
                        " 'connector' = 'datagen' ," +
                        " 'rows-per-second' = '1' , " +
                        " 'number-of-rows' = '100' , " +
                        " 'fields.id.kind' = 'random' , " +
                        " 'fields.id.length' = '5' , " +
                        " 'fields.vc.kind' = 'random' , " +
                        " 'fields.vc.min' = '100', " +
                        " 'fields.vc.max' = '200', " +
                        " 'fields.ts.kind' = 'sequence' , " +
                        " 'fields.ts.start' = '10000' , "  +
                        " 'fields.ts.end' = '11000'"  +
                        " )" ;
        //创建表
        tableEnv.executeSql(createSourceSql);

        //查询转换
        String selectSql = "select id,vc,ts from t1";
        Table table = tableEnv.sqlQuery(selectSql);
        //table.execute().print();
        tableEnv.createTemporaryView("t2",table);

        //使用print连接器将结果打印到控制台
        String  createSinkSql =
                " create table t3 (" +
                        " id STRING , " +
                        " vc INT , " +
                        " ts BIGINT " +
                        " ) WITH (" +
                        " 'connector' = 'print'" +
                        ")" ;
        tableEnv.executeSql(createSinkSql);
//        String insertSql = "insert into t3 select id,vc,ts from t2";
//        tableEnv.sqlQuery(insertSql).execute().print();

        tableEnv.executeSql("insert into t3 select id,vc,ts from t2");


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
